package com.atguigu.hadoop.examples.mapreduce;


import com.google.common.base.Charsets;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class BaileyBorweinPlouffe extends Configured
        implements Tool {
    public static final String DESCRIPTION = "A map/reduce program that uses Bailey-Borwein-Plouffe to compute exact digits of Pi.";
    private static final String NAME = new StringBuilder().append("mapreduce.")
            .append(BaileyBorweinPlouffe.class
                    .getSimpleName()).toString();

    private static final String WORKING_DIR_PROPERTY = new StringBuilder().append(NAME).append(".dir").toString();
    private static final String HEX_FILE_PROPERTY = new StringBuilder().append(NAME).append(".hex.file").toString();
    private static final String DIGIT_START_PROPERTY = new StringBuilder().append(NAME).append(".digit.start").toString();
    private static final String DIGIT_SIZE_PROPERTY = new StringBuilder().append(NAME).append(".digit.size").toString();
    private static final String DIGIT_PARTS_PROPERTY = new StringBuilder().append(NAME).append(".digit.parts").toString();

    private static final Logger LOG = LoggerFactory.getLogger(BaileyBorweinPlouffe.class);
    private static final long IMPLEMENTATION_LIMIT = 100000000L;
    private static final long ACCURACY_BIT = 32L;
    private static final long BBP_HEX_DIGITS = 4L;
    private static final long BBP_MULTIPLIER = 65536L;
    private static final long MAX_N = 4294967295L;

    private static <T> void print(PrintWriter out, Iterator<T> iterator, String prefix, String format, int elementsPerGroup, int groupsPerLine) {
        StringBuilder sb = new StringBuilder("\n");
        for (int i = 0; i < prefix.length(); i++)
            sb.append(" ");
        String spaces = sb.toString();

        out.print(new StringBuilder().append("\n").append(prefix).toString());
        for (int i = 0; iterator.hasNext(); i++) {
            if ((i > 0) && (i % elementsPerGroup == 0))
                out.print(i / elementsPerGroup % groupsPerLine == 0 ? spaces : " ");
            out.print(String.format(format, new Object[]{iterator.next()}));
        }
        out.println();
    }

    private static Job createJob(String name, Configuration conf)
            throws IOException {
        Job job = Job.getInstance(conf, new StringBuilder().append(NAME).append("_").append(name).toString());
        Configuration jobconf = job.getConfiguration();
        job.setJarByClass(BaileyBorweinPlouffe.class);

        job.setMapperClass(BbpMapper.class);
        job.setMapOutputKeyClass(LongWritable.class);
        job.setMapOutputValueClass(BytesWritable.class);

        job.setReducerClass(BbpReducer.class);
        job.setOutputKeyClass(LongWritable.class);
        job.setOutputValueClass(BytesWritable.class);
        job.setNumReduceTasks(1);

        job.setInputFormatClass(BbpInputFormat.class);

        jobconf.setLong("mapreduce.task.timeout", 0L);

        jobconf.setBoolean("mapreduce.map.speculative", false);
        jobconf.setBoolean("mapreduce.reduce.speculative", false);
        return job;
    }

    private static void compute(int startDigit, int nDigits, int nMaps, String workingDir, Configuration conf, PrintStream out)
            throws IOException {
        String name = new StringBuilder().append(startDigit).append("_").append(nDigits).toString();

        out.println(new StringBuilder().append("Working Directory = ").append(workingDir).toString());
        out.println();
        FileSystem fs = FileSystem.get(conf);
        Path dir = fs.makeQualified(new Path(workingDir));
        if (fs.exists(dir)) {
            throw new IOException(new StringBuilder().append("Working directory ").append(dir).append(" already exists.  Please remove it first.").toString());
        }
        if (!fs.mkdirs(dir)) {
            throw new IOException(new StringBuilder().append("Cannot create working directory ").append(dir).toString());
        }

        out.println(new StringBuilder().append("Start Digit      = ").append(startDigit).toString());
        out.println(new StringBuilder().append("Number of Digits = ").append(nDigits).toString());
        out.println(new StringBuilder().append("Number of Maps   = ").append(nMaps).toString());

        Job job = createJob(name, conf);
        Path hexfile = new Path(dir, new StringBuilder().append("pi_").append(name).append(".hex").toString());
        FileOutputFormat.setOutputPath(job, new Path(dir, "out"));

        job.getConfiguration().set(WORKING_DIR_PROPERTY, dir.toString());
        job.getConfiguration().set(HEX_FILE_PROPERTY, hexfile.toString());

        job.getConfiguration().setInt(DIGIT_START_PROPERTY, startDigit);
        job.getConfiguration().setInt(DIGIT_SIZE_PROPERTY, nDigits);
        job.getConfiguration().setInt(DIGIT_PARTS_PROPERTY, nMaps);

        out.println("\nStarting Job ...");
        long startTime = Time.monotonicNow();
        try {
            if (!job.waitForCompletion(true)) {
                out.println("Job failed.");
                System.exit(1);
            }
        } catch (Exception e) {
            double duration;
            throw new RuntimeException(e);
        } finally {
            double duration = (Time.monotonicNow() - startTime) / 1000.0D;
            out.println(new StringBuilder().append("Duration is ").append(duration).append(" seconds.").toString());
        }
        out.println(new StringBuilder().append("Output file: ").append(hexfile).toString());
    }

    public int run(String[] args)
            throws IOException {
        if (args.length != 4) {
            System.err.println("Usage: bbp  <startDigit> <nDigits> <nMaps> <workingDir>");

            ToolRunner.printGenericCommandUsage(System.err);
            return -1;
        }

        int startDigit = Integer.parseInt(args[0]);
        int nDigits = Integer.parseInt(args[1]);
        int nMaps = Integer.parseInt(args[2]);
        String workingDir = args[3];

        if (startDigit <= 0)
            throw new IllegalArgumentException(new StringBuilder().append("startDigit = ").append(startDigit).append(" <= 0").toString());
        if (nDigits <= 0)
            throw new IllegalArgumentException(new StringBuilder().append("nDigits = ").append(nDigits).append(" <= 0").toString());
        if (nDigits % 4L != 0L) {
            throw new IllegalArgumentException(new StringBuilder().append("nDigits = ").append(nDigits).append(" is not a multiple of ").append(4L).toString());
        }
        if (nDigits - 1L + startDigit > 100000004L) {
            throw new UnsupportedOperationException(new StringBuilder().append("nDigits - 1 + startDigit = ").append(nDigits - 1L + startDigit).append(" > IMPLEMENTATION_LIMIT + BBP_HEX_DIGITS,, where IMPLEMENTATION_LIMIT=").append(100000000L).append("and BBP_HEX_DIGITS=").append(4L).toString());
        }

        if (nMaps <= 0) {
            throw new IllegalArgumentException(new StringBuilder().append("nMaps = ").append(nMaps).append(" <= 0").toString());
        }

        compute(startDigit, nDigits, nMaps, workingDir, getConf(), System.out);
        return 0;
    }

    public static void main(String[] argv) throws Exception {
        System.exit(ToolRunner.run(null, new BaileyBorweinPlouffe(), argv));
    }

    static long hexDigits(long d) {
        if (d < 0L)
            throw new IllegalArgumentException(new StringBuilder().append("d = ").append(d).append(" < 0").toString());
        if (d > 100000000L) {
            throw new IllegalArgumentException(new StringBuilder().append("d = ").append(d).append(" > IMPLEMENTATION_LIMIT = ").append(100000000L).toString());
        }

        double s1 = sum(1L, d);
        double s4 = sum(4L, d);
        double s5 = sum(5L, d);
        double s6 = sum(6L, d);

        double pi = s1 + s1;
        if (pi >= 1.0D)
            pi -= 1.0D;
        pi *= 2.0D;
        if (pi >= 1.0D) {
            pi -= 1.0D;
        }
        pi -= s4;
        if (pi < 0.0D)
            pi += 1.0D;
        pi -= s4;
        if (pi < 0.0D)
            pi += 1.0D;
        pi -= s5;
        if (pi < 0.0D)
            pi += 1.0D;
        pi -= s6;
        if (pi < 0.0D) {
            pi += 1.0D;
        }
        return (long) (pi * 65536.0D);
    }

    private static double sum(long j, long d) {
        long k = j == 1L ? 1L : 0L;
        double s = 0.0D;
        if (k <= d) {
            s = 1.0D / (d << 3 | j);
            for (; k < d; k += 1L) {
                long n = k << 3 | j;
                s += mod(d - k << 2, n) * 1.0D / n;
                if (s >= 1.0D)
                    s -= 1.0D;
            }
            k += 1L;
        }

        if (k >= 33554432L) {
            return s;
        }
        for (; ; k += 1L) {
            long n = k << 3 | j;
            long shift = k - d << 2;
            if ((32L <= shift) || (1L << (int) (32L - shift) < n)) {
                return s;
            }

            s += 1.0D / (n << (int) shift);
            if (s >= 1.0D)
                s -= 1.0D;
        }
    }

    static long mod(long e, long n) {
        long mask = (e & 0x0) == 0L ? 4294967295L : -4294967296L;

        mask &= ((e & 0xFFFF0000 & mask) == 0L ? 281470681808895L : -281470681808896L);

        mask &= ((e & 0xFF00FF00 & mask) == 0L ? 71777214294589695L : -71777214294589696L);

        mask &= ((e & 0xF0F0F0F0 & mask) == 0L ? 1085102592571150095L : -1085102592571150096L);

        mask &= ((e & 0xCCCCCCCC & mask) == 0L ? 3689348814741910323L : -3689348814741910324L);

        mask &= ((e & 0xAAAAAAAA & mask) == 0L ? 6148914691236517205L : -6148914691236517206L);

        long r = 2L;
        for (mask >>= 1; mask > 0L; mask >>= 1) {
            r *= r;
            r %= n;

            if ((e & mask) != 0L) {
                r += r;
                if (r >= n)
                    r -= n;
            }
        }
        return r;
    }

    static int[] partition(int offset, int size, int nParts) {
        int[] parts = new int[nParts];
        long total = workload(offset, size);
        int remainder = offset % 4;

        parts[0] = offset;
        for (int i = 1; i < nParts; i++) {
            long target = offset + i * (total / nParts) + i * (total % nParts) / nParts;

            int low = parts[(i - 1)];
            int high = offset + size;
            while (high > low + 4) {
                int mid = (high + low - 2 * remainder) / 8 * 4 + remainder;
                long midvalue = workload(mid);
                if (midvalue == target)
                    high = low = mid;
                else if (midvalue > target)
                    high = mid;
                else
                    low = mid;
            }
            parts[i] =
                    (workload(high) -
                            target > target - workload(low) ? low : high == low ? high :
                            high);
        }

        return parts;
    }

    private static long workload(long n) {
        if (n < 0L)
            throw new IllegalArgumentException(new StringBuilder().append("n = ").append(n).append(" < 0").toString());
        if (n > 4294967295L) {
            throw new IllegalArgumentException(new StringBuilder().append("n = ").append(n).append(" > MAX_N = ").append(4294967295L).toString());
        }
        return (n & 1L) == 0L ? (n >> 1) * (n + 1L) : n * (n + 1L >> 1);
    }

    private static long workload(long offset, long size) {
        return workload(offset + size) - workload(offset);
    }

    private static class Fraction {
        private final int[] integers;
        private int first = 0;

        Fraction(List<Byte> bytes) {
            this.integers = new int[(bytes.size() + 2) / 3];
            for (int i = 0; i < bytes.size(); i++) {
                int b = 0xFF & ((Byte) bytes.get(i)).byteValue();
                this.integers[(this.integers.length - 1 - i / 3)] |= b << (2 - i % 3 << 3);
            }
            skipZeros();
        }

        int times10() {
            int carry = 0;
            for (int i = this.first; i < this.integers.length; i++) {
                this.integers[i] <<= 1;
                this.integers[i] += carry + (this.integers[i] << 2);
                carry = this.integers[i] >> 24;
                this.integers[i] &= 16777215;
            }
            skipZeros();
            return carry;
        }

        private void skipZeros() {
            while ((this.first < this.integers.length) && (this.integers[this.first] == 0)) this.first += 1;
        }
    }

    public static class BbpInputFormat extends InputFormat<LongWritable, IntWritable> {
        public List<InputSplit> getSplits(JobContext context) {
            int startDigit = context.getConfiguration().getInt(
                    BaileyBorweinPlouffe.DIGIT_START_PROPERTY, 1);
            int nDigits = context.getConfiguration().getInt(
                    BaileyBorweinPlouffe.DIGIT_SIZE_PROPERTY, 100);
            int nMaps = context.getConfiguration().getInt(
                    BaileyBorweinPlouffe.DIGIT_PARTS_PROPERTY, 1);

            List splits = new ArrayList(nMaps);
            int[] parts = BaileyBorweinPlouffe.partition(startDigit - 1, nDigits, nMaps);
            for (int i = 0; i < parts.length; i++) {
                int k = i < parts.length - 1 ? parts[(i + 1)] : nDigits + startDigit - 1;
                splits.add(new BaileyBorweinPlouffe.BbpSplit(i, parts[i], k - parts[i]));
            }
            return splits;
        }

        public RecordReader<LongWritable, IntWritable> createRecordReader(InputSplit generic, TaskAttemptContext context) {
            final BaileyBorweinPlouffe.BbpSplit split = (BaileyBorweinPlouffe.BbpSplit) generic;

            return new RecordReader() {
                boolean done = false;

                public void initialize(InputSplit split, TaskAttemptContext context) {
                }

                public boolean nextKeyValue() {
                    return !this.done ? (this.done = true) : false;
                }

                public LongWritable getCurrentKey() {
                    return new LongWritable();
                }

                public IntWritable getCurrentValue() {
                    return new IntWritable((int) split.getLength());
                }

                public float getProgress() {
                    return this.done ? 1.0F : 0.0F;
                }

                public void close() {
                }
            };
        }
    }

    public static class BbpSplit extends InputSplit
            implements Writable {
        private static final String[] EMPTY = new String[0];
        private long offset;
        private int size;

        public BbpSplit() {
        }

        private BbpSplit(int i, long offset, int size) {
            BaileyBorweinPlouffe.LOG.info("Map #" + i + ": workload=" + BaileyBorweinPlouffe.workload(offset, size) + ", offset=" + offset + ", size=" + size);

            this.offset = offset;
            this.size = size;
        }

        private long getOffset() {
            return this.offset;
        }

        public long getLength() {
            return this.size;
        }

        public String[] getLocations() {
            return EMPTY;
        }

        public void readFields(DataInput in) throws IOException {
            this.offset = in.readLong();
            this.size = in.readInt();
        }

        public void write(DataOutput out) throws IOException {
            out.writeLong(this.offset);
            out.writeInt(this.size);
        }
    }

    public static class BbpReducer extends Reducer<LongWritable, BytesWritable, LongWritable, BytesWritable> {
        private final List<Byte> hex = new ArrayList();

        protected void reduce(LongWritable offset, Iterable<BytesWritable> values, Reducer<LongWritable, BytesWritable, LongWritable, BytesWritable>.Context context)
                throws IOException, InterruptedException {
            for (BytesWritable bytes : values) {
                for (int i = 0; i < bytes.getLength(); i++) {
                    this.hex.add(Byte.valueOf(bytes.getBytes()[i]));
                }
            }
            BaileyBorweinPlouffe.LOG.info("hex.size() = " + this.hex.size());
        }

        protected void cleanup(Reducer<LongWritable, BytesWritable, LongWritable, BytesWritable>.Context context)
                throws IOException, InterruptedException {
            Configuration conf = context.getConfiguration();
            Path dir = new Path(conf.get(BaileyBorweinPlouffe.WORKING_DIR_PROPERTY));
            FileSystem fs = dir.getFileSystem(conf);

            Path hexfile = new Path(conf.get(BaileyBorweinPlouffe.HEX_FILE_PROPERTY));
            OutputStream out = new BufferedOutputStream(fs.create(hexfile));
            try {
                Iterator<Byte> iterator = this.hex.iterator();
                while (iterator.hasNext()) {
                    Byte next = iterator.next();
                    out.write(next.byteValue());
                }

            } finally {
                Iterator localIterator;
                out.close();
            }

            if (conf.getInt(BaileyBorweinPlouffe.DIGIT_START_PROPERTY, 1) == 1) {
                Path outfile = new Path(dir, "pi.txt");
                BaileyBorweinPlouffe.LOG.info("Writing text output to " + outfile);
                OutputStream outputstream = fs.create(outfile);
                try {
                    PrintWriter outs = new PrintWriter(new OutputStreamWriter(outputstream, Charsets.UTF_8), true);

                    BaileyBorweinPlouffe.print(outs, this.hex.iterator(), "Pi = 0x3.", "%02X", 5, 5);
                    outs.println("Total number of hexadecimal digits is " + 2 * this.hex
                            .size() + ".");

                    final BaileyBorweinPlouffe.Fraction dec = new BaileyBorweinPlouffe.Fraction(this.hex);
                    final int decDigits = 2 * this.hex.size();
                    BaileyBorweinPlouffe.print(outs, new Iterator() {
                                private int i = 0;

                                public boolean hasNext() {
                                    return this.i < decDigits;
                                }

                                public Integer next() {
                                    this.i += 1;
                                    return Integer.valueOf(dec.times10());
                                }

                                public void remove() {
                                }
                            }
                            , "Pi = 3.", "%d", 10, 5);

                    outs.println("Total number of decimal digits is " + decDigits + ".");
                } finally {
                    outputstream.close();
                }
            }
        }
    }

    public static class BbpMapper extends Mapper<LongWritable, IntWritable, LongWritable, BytesWritable> {
        protected void map(LongWritable offset, IntWritable length, Mapper<LongWritable, IntWritable, LongWritable, BytesWritable>.Context context)
                throws IOException, InterruptedException {
            BaileyBorweinPlouffe.LOG.info("offset=" + offset + ", length=" + length);

            byte[] bytes = new byte[length.get() >> 1];
            long d = offset.get();
            for (int i = 0; i < bytes.length; d += 4L) {
                long digits = BaileyBorweinPlouffe.hexDigits(d);
                bytes[(i++)] = ((byte) (int) (digits >> 8));
                bytes[(i++)] = ((byte) (int) digits);
            }

            context.write(offset, new BytesWritable(bytes));
        }
    }
}